#include "config.h"

#include <sys/types.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>

#define DBG_SUBSYS S_LIBCONTROL

#include "limits.h"
#include "adt.h"
#include "ynet_rpc.h"
#include "sysy_lib.h"
#include "cluster.h"
#include "chunk.h"
#include "bmap.h"
#include "metadata.h"
#include "net_table.h"
#include "configure.h"
#include "net_global.h"
#include "stor_ctl.h"
#include "cache.h"
#include "pool_proto.h"
#include "volume_proto.h"
#include "volume_ctl.h"
#include "volume_bh.h"
#include "../chunk/chunk_cleanup.h"
#include "../../storage/task/recovery.h"
#include "../../storage/task/balance2.h"
#include "pool_ctl.h"
#include "ylog.h"
#include "dbg.h"

int stor_ctl_mkvol(const chkid_t *parent, const char *name, const char *site_name, const setattr_t *setattr,
                  chkinfo_t *chkinfo)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_mkvol(parent, name, site_name, setattr, chkinfo);
        else
                return ENOTDIR;
}

int stor_ctl_mkvolwith(const chkid_t *parent, const char *name, const chkinfo_t *chkinfo)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_mkvolwith(parent, name, chkinfo);
        else
                return ENOTDIR;
}

int stor_ctl_mkpool(const chkid_t *parent, const char *name, const char *site_name,
                const setattr_t *setattr, chkinfo_t *chkinfo)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_mkpool(parent, name, site_name, setattr, chkinfo);
        else
                return ENOTDIR;
}

int stor_ctl_extend_pool(const chkid_t *poolid)
{
        if (poolid->type == __POOL_CHUNK__)
                return pool_ctl_extend(poolid);
        else
                return ENOTDIR;
}

int stor_ctl_listpool_open(const chkid_t *parent, const char *uuid)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_listpool_open(parent, uuid);
        else
                return ENOTDIR;
}

int stor_ctl_listpool(const chkid_t *parent, const char *uuid, uint64_t offset, void *de, int *delen)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_listpool(parent, uuid, offset, de, delen);
        else
                return ENOTDIR;
}

int stor_ctl_listpool_close(const chkid_t *parent, const char *uuid)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_listpool_close(parent, uuid);
        else
                return ENOTDIR;
}

int stor_ctl_lookup_srv(const chkid_t *chkid, const fileid_t *parent)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_lookup_srv(chkid, parent);
        else
                return volume_ctl_lookup_srv(chkid, parent);

}

int stor_ctl_lookup(const chkid_t *parent, const char *name,
                  chkinfo_t *chkinfo)
{
        if (parent->type == __POOL_CHUNK__) {
                return pool_ctl_lookup(parent, name, chkinfo);
        } else {
                return volume_ctl_snapshot_lookup(parent, name, chkinfo);//snapshot lookup
        }
}

int stor_ctl_chunk_getinfo(const chkid_t *parent, const chkid_t *chkid, chkinfo_t *chkinfo)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_chunk_getinfo(parent, chkid, chkinfo);
        else
                return volume_ctl_chunk_getinfo(parent, chkid, chkinfo);
}

int stor_ctl_getpool(const chkid_t *chkid, char *pool)
{
        if (chkid->type == __POOL_CHUNK__)
                return EISDIR;
        else
                return volume_ctl_getpool(chkid, pool);
}

int stor_ctl_getattr(const chkid_t *chkid, fileinfo_t *fileinfo)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_getattr(chkid, fileinfo);
        else
                return volume_ctl_getattr(chkid, fileinfo);
}

int stor_ctl_setattr(const chkid_t *chkid, fileinfo_t *fileinfo, const setattr_t *setattr)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_setattr(chkid, fileinfo, setattr);
        else
                return volume_ctl_setattr(chkid, fileinfo, setattr);
}

int stor_ctl_chunk_allocate(const fileid_t *fileid,
                const chkid_t *chkid, int chknum, int fill)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_chunk_allocate(fileid, chkid, chknum, fill);
        else
                return EISDIR;
}

int stor_ctl_chunk_migrate(const fileid_t *parent, const chkid_t *chkid, uint32_t force,
                          chkinfo_t *chkinfo)
{
        if (chkid->type == __VOLUME_CHUNK__)
                return volume_ctl_chunk_migrate(parent, chkid, force, chkinfo);
        else
                return pool_ctl_chunk_migrate(parent, chkid, force, chkinfo);
}

int stor_ctl_chunk_reject(const fileid_t *parent, const chkid_t *chkid, const nid_t *bad,
                          chkinfo_t *chkinfo)
{
        if (parent->type == __VOLUME_CHUNK__)
                return volume_ctl_chunk_reject(parent, chkid, bad, chkinfo);
        else
                return pool_ctl_chunk_reject(parent, chkid, bad, chkinfo);
}

int stor_ctl_chunk_update(const fileid_t *parent, const chkinfo_t *chkinfo, const nid_t *owner, uint64_t info_version)
{
        YASSERT(chkinfo->repnum <= LICH_REPLICA_MAX);

        if (parent->type == __POOL_CHUNK__) {
                return pool_ctl_chunk_update(parent, chkinfo, owner, info_version);
        } else {
                return volume_ctl_chunk_update(parent, chkinfo, owner, info_version);
        }
}

int stor_ctl_chunk_cleanup(const fileid_t *parent, const chkid_t *chkid, const nid_t *nid, uint64_t meta_version)
{
        if (parent->type == __VOLUME_CHUNK__) {
                return volume_ctl_chunk_cleanup(parent, chkid, nid, meta_version);
        } else {
                return pool_ctl_chunk_cleanup(parent, chkid, nid, meta_version);
        }
}

int stor_ctl_chunk_set(const fileid_t *parent, const chkid_t *chkid,
                       const nid_t *nid, int status)
{
        if (parent->type == __POOL_CHUNK__)
                return pool_ctl_chunk_set(parent, chkid, nid, status);
        else
                return volume_ctl_chunk_set(parent, chkid, nid, status);
}

int stor_ctl_vfm_set(const fileid_t *parent, const chkid_t *chkid,
                     const vfm_t *vfm)
{
        return volume_ctl_vfm_set(parent, chkid, vfm);
}

int stor_ctl_vfm_get(const fileid_t *parent, const chkid_t *chkid,
                     vfm_t *vfm)
{
        return volume_ctl_vfm_get(parent, chkid, vfm);
}

int stor_ctl_rmpool(const chkid_t *chkid, const char *name)
{
        return pool_ctl_rmpool(chkid, name);
}

int stor_ctl_rmvol(const chkid_t *chkid, const char *name, int force)
{
        return pool_ctl_unlink(chkid, name, force);
}

int stor_ctl_cleanup(const chkid_t *chkid, const char *name)
{
        if (chkid->type == __VOLUME_CHUNK__)
                return volume_ctl_cleanup(chkid);
        else
                return pool_ctl_cleanup(chkid, name);
}

int stor_ctl_cleanup_bh(const chkid_t *chkid, const char *name)
{
        if (chkid->type == __POOL_CHUNK__)
                return volume_bh_filecleanup(chkid, name);
        else
                return ENOTDIR;
}

int stor_ctl_rollback_bh(const chkid_t *chkid)
{
        if (chkid->type == __VOLUME_CHUNK__)
                return volume_bh_add(chkid, BH_SNAP_ROLLBACK, 1);
        else
                return EISDIR;
}

int stor_ctl_flat_bh(const chkid_t *chkid)
{
        if (chkid->type == __VOLUME_CHUNK__)
                return volume_bh_add(chkid, BH_SNAP_FLAT, 1);
        else
                return EISDIR;
}

int stor_ctl_rmsnap_bh(const chkid_t *chkid, const char *name)
{
        if (chkid->type == __VOLUME_CHUNK__)
                return volume_bh_snapcleanup(chkid, name);
        else
                return EISDIR;
}

int stor_ctl_rename_lock(const fileid_t *fileid, const fileid_t *src, const char *name, int force)
{
        return pool_ctl_rename_lock(fileid, src, name, force);
}

int stor_ctl_rename_unlock(const fileid_t *fileid, const chkinfo_t *chkinfo)
{
        return pool_ctl_rename_unlock(fileid, chkinfo);
}

int stor_ctl_rename(const fileid_t *from, const char *fname,
                  const fileid_t *to, const char *tname)
{
        return pool_ctl_rename(from, fname, to, tname);
}

/** 设定卷的chunk副本位置
 *
 * @param srv  volume chkid
 * @param chkid chkid
 * @param dist  target reploc
 * @param dist_count
 * @return
 */
int stor_ctl_chunk_move(const fileid_t *srv, const chkid_t *chkid, const nid_t *dist, int dist_count)
{
        int i;

        metadata_check_diskid(dist, dist_count);
        DBUG("srv "CHKID_FORMAT", chkid: "CHKID_FORMAT", dist_count: %d\n", CHKID_ARG(srv), CHKID_ARG(chkid), dist_count);
        for (i = 0; i < dist_count; i++) {
                DBUG("chunk move to %s\n", network_rname(&dist[i]));
        }

        if (chkid->type == __POOL_CHUNK__ || chkid->type == __POOL_SUB_CHUNK__) {
                return pool_ctl_chunk_move(srv, chkid, dist, dist_count);
        } else if (chkid->type == __VOLUME_CHUNK__ || chkid->type ==  __VOLUME_SUB_CHUNK__ || chkid->type == __RAW_CHUNK__) {
                return volume_ctl_chunk_move(srv, chkid, dist, dist_count);
        } else {
                YASSERT(0);
        }

        return 0;
}

int stor_ctl_chunk_check(const fileid_t *parent, const chkid_t *chkid, int idx, int async, int force, int *oflags)
{
        (void) idx;

        if (async) {
                if (chkid->type == __POOL_CHUNK__ || chkid->type == __POOL_SUB_CHUNK__) {
                        return pool_ctl_chunk_check(parent, chkid);
                } else if (chkid->type == __VOLUME_CHUNK__) {
                        return volume_ctl_chunk_check(chkid, chkid);
                } else {
                        return volume_ctl_chunk_check(parent, chkid);
                }
        } else {
                if (chkid->type == __POOL_CHUNK__ || chkid->type == __POOL_SUB_CHUNK__) {
                        return pool_ctl_chunk_sync(parent, chkid);
                } else if (chkid->type == __VOLUME_CHUNK__) {
                        return volume_ctl_chunk_sync(chkid, chkid, force, oflags);
                } else {
                        return volume_ctl_chunk_sync(parent, chkid, force, oflags);
                }
        }
}

int stor_ctl_chunk_check_multi(const fileid_t *parent, const chkid_t *chkid, int chknum, nid_t *srcnid, int *retval)
{
        int ret, count;
        chkid_t *id;

        if (parent->type == __VOLUME_CHUNK__) {
                return volume_ctl_chunk_sync_force(srcnid, parent, chkid, chknum, retval);
        } else {
                for (count = 0; count < chknum; count ++) {
                        id = (chkid_t *)&chkid[count];
                        ret = pool_ctl_chunk_sync_force(parent, id, srcnid);
                        if (unlikely(ret)) {
                                DWARN(""CHKID_FORMAT" check failed, ret %d\n", CHKID_ARG(id), ret);
                                retval[count] = ret;
                        }
                }
        }

        return 0;
}

int stor_ctl_xattr_remove(const chkid_t *chkid, const char *key)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_xattr_remove(chkid, key);
        else
                return volume_ctl_xattr_remove(chkid, key);
}

int stor_ctl_move(const chkid_t *chkid, const nid_t *nid, int count)
{
        metadata_check_diskid(nid, count);
        YASSERT(chkid->type == __VOLUME_CHUNK__);
        return volume_ctl_move(chkid, nid, count);
}

int stor_ctl_localize(const chkid_t *chkid, int idx)
{
        YASSERT(chkid->type == __VOLUME_CHUNK__);
        return volume_ctl_localize(chkid, idx);
}

int stor_ctl_stat(const fileid_t *fileid, filestat_t *filestat, off_t off, size_t size)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_stat(fileid, filestat, off, size);
        else
                return EISDIR;
}

int stor_ctl_check_ready(const fileid_t *fileid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_check_ready(fileid);
}

int stor_ctl_group_snapshot_lock(const fileid_t *fileid, uuid_t *_uuid)
{
        return volume_ctl_snapshot_wrlock(fileid, _uuid);
}

int stor_ctl_group_snapshot_unlock(const fileid_t *fileid, const uuid_t uuid)
{
        return volume_ctl_snapshot_unlock(fileid, uuid);
}

int stor_ctl_group_snapshot_create_nolock(const fileid_t *fileid, const char *name, const char *_site, const uuid_t uuid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_create_nolock(fileid, name, _site, uuid);
}

int stor_ctl_group_snapshot_remove_nolock(const fileid_t *fileid, const char *name, const uuid_t uuid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_remove_nolock(fileid, name, uuid);
}

int stor_ctl_snapshot_check(const fileid_t *fileid, const char *name)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_check(fileid, name);
}

int stor_ctl_snapshot_isempty(const fileid_t *fileid, int *empty)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_isempty(fileid, empty);
}

int stor_ctl_table_read(const fileid_t *fileid, const chkid_t *chkid, buffer_t *buf, uint32_t size, uint64_t offset)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_table_read(fileid, chkid, buf, size, offset);
        else
                return EISDIR;
}

int stor_ctl_table_write(const fileid_t *fileid, const chkid_t *chkid, const buffer_t *buf, uint32_t size, uint64_t offset)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_table_write(fileid, chkid, buf, size, offset);
        else
                return EISDIR;
}

int stor_ctl_connection(const fileid_t *fileid, const nid_t *nid, void *list, int *count)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_connection(fileid, nid, list, count);
        else
                return EISDIR;
}

int stor_ctl_vfm_cleanup(const fileid_t *fileid, const chkid_t *tid)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_vfm_cleanup(fileid, tid);
        else
                return EISDIR;
}

int stor_ctl_vfm_stat(const fileid_t *fileid, int *count)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_vfm_stat(fileid, count);
        else
                return EISDIR;
}

int stor_ctl_connect(const fileid_t *fileid, const nid_t *nid, const char *addr)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_connect(fileid, nid, addr);
        else
                return EISDIR;
}

int stor_ctl_disconnect(const fileid_t *fileid, const nid_t *nid, const char *addr)
{
        if (fileid->type == __VOLUME_CHUNK__)
                return volume_ctl_disconnect(fileid, nid, addr);
        else
                return EISDIR;
}

int stor_ctl_snapshot_create(const fileid_t *fileid, const char *name, int p, const char *_site)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_create(fileid, name, p, _site);
}

int stor_ctl_snapshot_rollback(const fileid_t *fileid, const char *name)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_rollback(fileid, name);
}

int stor_ctl_snapshot_flat(const fileid_t *fileid, const int idx, int force)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_flat(fileid, idx, force);
}

int stor_ctl_snapshot_remove(const fileid_t *fileid, const char *name, int force)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_remove(fileid, name, force);
}

int stor_ctl_snapshot_protect(const fileid_t *fileid, const snap_protect_param_t on)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_protect(fileid, on);
}

int stor_ctl_snapshot_updateparent(const fileid_t *fileid, const char *name, const uint64_t from)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_updateparent(fileid, name, from);
}

int stor_ctl_snapshot_setfrom(const fileid_t *fileid, const uint64_t from)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_setfrom(fileid, from);
}

int stor_ctl_snapshot_last(const fileid_t *fileid, nid_t *snapnid, fileid_t *snapid, char *snapname, uint64_t *snap_version)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_last(fileid, snapnid, snapid, snapname, snap_version);
}

int stor_ctl_snapshot_prev(const fileid_t *fileid, const fileid_t *snapid, fileid_t *previd, char *name, uint64_t *snap_version)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_prev(fileid, snapid, previd, name, snap_version);
}

#if 0
int stor_ctl_snapshot_cleanup(const fileid_t *fileid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_cleanup(fileid);
}
#endif

int stor_ctl_snapshot_listopen(const chkid_t *fileid, const char *uuid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_listopen(fileid, uuid);
}

int stor_ctl_snapshot_list(const chkid_t *parent, const char *uuid, uint64_t offset, void *de, int *delen)
{
        YASSERT(parent->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_list(parent, uuid, offset, de, delen);
}

int stor_ctl_snapshot_listclose(const chkid_t *fileid, const char *uuid)
{
        YASSERT(fileid->type == __VOLUME_CHUNK__);
        return volume_ctl_snapshot_listclose(fileid, uuid);
}

int stor_ctl_xattr_get(const chkid_t *chkid, const char *key, char *value, int *valuelen)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_xattr_get(chkid, key, value, valuelen);
        else
                return volume_ctl_xattr_get(chkid, key, value, valuelen);
}

int stor_ctl_xattr_set(const chkid_t *chkid, const char *key, const char *value,
                    uint32_t valuelen, int flag)
{
        YASSERT(strcmp(key, STORAGE_AREA_KEY));
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_xattr_set(chkid, key, value, valuelen, flag);
        else
                return volume_ctl_xattr_set(chkid, key, value, valuelen, flag);
}

int stor_ctl_xattr_list(const chkid_t *chkid, char *buf, int *buflen)
{
        if (chkid->type == __POOL_CHUNK__)
                return pool_ctl_xattr_list(chkid, buf, buflen);
        else
                return volume_ctl_xattr_list(chkid, buf, buflen);
}

int stor_ctl_init(uint64_t max_chunk)
{
        int ret;

        // sync, localize
        ret = chunk_proto_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = volume_ctl_init(max_chunk);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = pool_ctl_init(max_chunk);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        // 离线消息处理
        ret = chunk_cleanup_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        // 后台任务: 包括vol cleanup, rmsnap, rollback and flat
        ret = volume_bh_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        // 恢复线程
        ret = recovery_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

#if ENABLE_BALANCE
        // 数据平衡线程
#if 0
        ret = balance_init();
#else
        ret = balance_pool_init();
#endif
        if (unlikely(ret))
                GOTO(err_ret, ret);
#endif

        return 0;
err_ret:
        return ret;
}

int stor_ctl_chunk_exist(const fileid_t *fileid, const chkid_t *chkid, int *exist) {
        if (chkid->type == __VOLUME_CHUNK__) {
                return volume_ctl_chunk_exist(fileid, chkid, exist);
        } else {
                return EISDIR;
        }
}
